package com.hsd.znsh.config;

import com.alibaba.fastjson.JSON;
import com.hsd.znsh.entity.*;
import com.hsd.znsh.enums.AckEnum;
import com.hsd.znsh.enums.OnlineStatusEnum;
import com.hsd.znsh.enums.ReadStatusEnum;
import com.hsd.znsh.properties.MqttProperties;
import com.hsd.znsh.repository.*;
import com.hsd.znsh.sensor.SensorDataList;
import com.hsd.znsh.sensor.SensorKeyValue;
import com.hsd.znsh.service.PushService;
import com.hsd.znsh.utils.DateUtil;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.ExecutorService;


@Configuration
@EnableConfigurationProperties({MqttProperties.class})
public class ZnshMqttConfig {

    @Autowired
    private MqttProperties mqttProperties;

    @Autowired
    private SensorRepository sensorRepository;

    @Autowired
    private SensorDataRepository sensorDataRepository;

    @Autowired
    private SensorTypePropertyRepository sensorTypePropertyRepository;

    @Autowired
    private PropertyRepository propertyRepository;

    @Autowired
    private PushService pushService;

    @Autowired
    private HostRepository hostRepository;

    @Autowired
    private EventRepository eventRepository;

    @Autowired
    private ExecutorService executorService;

    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageProducer inbound() {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getClientId1(), mqttClientFactory());
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setOutputChannel(mqttInputChannel());
        for (String top : mqttProperties.getTopics()) {
            adapter.addTopic(top, 1);
        }
        return adapter;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                String topic = message.getHeaders().get("mqtt_topic", String.class);
                System.out.println(topic);
                Task task = new Task(message.getPayload().toString());
                executorService.execute(task);
            }
        };
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler =
                new MqttPahoMessageHandler(mqttProperties.getClientId2(), mqttClientFactory());
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic(mqttProperties.getDefaultTopic());
        return messageHandler;
    }

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setServerURIs(mqttProperties.getServer());
        factory.setUserName(mqttProperties.getUsername());
        factory.setPassword(mqttProperties.getPassword());
        return factory;
    }

    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    @MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
    public interface MyGateway {
        void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String data);
    }

    private class Task implements Runnable {

        private String payLoad;

        public Task( String payLoad) {
            this.payLoad = payLoad;
        }

        @Override
        public void run() {
            Map map = JSON.parseObject(payLoad, Map.class);
            String data = (String) map.get("data");
            String jsonString = new String(Base64.getDecoder().decode(data));
            try {
                System.out.println(jsonString);
                SensorDataList sensorDataList = JSON.parseObject(jsonString, SensorDataList.class);
                saveSensorData(sensorDataList, new Date());
            } catch (Exception e) {}
        }

        @Transactional
        public void saveSensorData(SensorDataList sensorDataList, Date date) throws Exception {
            Sensor sensor = sensorRepository.findBySensorSn(sensorDataList.getSensorSn());
            List<SensorData> list = new ArrayList<>();
            if (sensor == null) return;

            List<Integer> propertyIds = sensorTypePropertyRepository.findPropertyIdListBySensorTypeId(
                    sensor.getSensorTypeId());
            int bh = 1;
            for (SensorKeyValue sensorKeyValue : sensorDataList.getData()) {
                Property property;
                if (propertyIds == null || propertyIds.size() == 0) {
                    property = new Property();
                    property.setName(sensorKeyValue.getName());
                    property.setUnit("");
                    propertyRepository.save(property);

                    SensorTypeProperty sensorTypeProperty = new SensorTypeProperty();
                    sensorTypeProperty.setSensorTypeId(sensor.getSensorTypeId());
                    sensorTypeProperty.setPropertyId(property.getId());
                    sensorTypeProperty.setBh(bh++);
                    sensorTypePropertyRepository.save(sensorTypeProperty);
                } else {
                    property = propertyRepository.findByNameAndIdIn(sensorKeyValue.getName(), propertyIds);
                    if (property == null) throw new Exception(sensorKeyValue.getName() + "不存在");
                }
                SensorData sensorData = new SensorData();
                sensorData.setTime(date);
                sensorData.setPropertyId(property.getId());
                sensorData.setValue(sensorKeyValue.getValue());
                sensorData.setSensorSn(sensorDataList.getSensorSn());
                list.add(sensorData);
            }
            push(sensor, date);
            sensor.setTime(date);
            sensor.setOnlineStatus(OnlineStatusEnum.ONLINE.getCode());
            sensorDataRepository.save(list);
            sensorRepository.save(sensor);
        }

        private void push(Sensor sensor, Date date) {
            LocalDateTime sensorTime = DateUtil.dateToLocalDateTime(sensor.getTime());
            LocalDateTime nowTime = DateUtil.dateToLocalDateTime(date);
            if (sensorTime.plusSeconds(sensor.getFrequency() * 2).isBefore(nowTime)) {
                System.out.println("设备上线" + DateUtil.format(date, "yyyy-MM-dd HH:mm:ss"));
                Host host = hostRepository.findOne(sensor.getHostId());
                String content;
                String title;
                if (sensor.getAlias() == null || sensor.getAlias().equals("")) {
                    content = sensor.getName() + "(" + sensor.getSensorSn() + ") 上线";
                } else {
                    content = sensor.getAlias() + "(" + sensor.getSensorSn() + ") 上线";
                }
                if (host.getAlias() == null || host.getAlias().equals("")) {
                    title = host.getHostName();
                } else {
                    title = host.getAlias();
                }
                Event event = new Event();
                event.setReadStatus(ReadStatusEnum.NOT_READ.getCode());
                event.setContent(content);
                event.setDeviceSn(sensor.getSensorSn());
                event.setEventLevel(1);
                event.setGatewayName(title);
                event.setEventType("通知");
                event.setGatewaySn(host.getHostSn());

                eventRepository.save(event);
                pushService.pushToGroup(title, content, host.getHostSn());
            }
        }
    }


}
